package com.baishancloud.log.douyin.demand1.subtask

import com.baishancloud.log.common.entity.ServerCacheGroupEntity
import com.baishancloud.log.common.util.{ExternalDataFetchUtil, IpInfoUtil, StringUtil, TimeUtil}
import com.baishancloud.log.douyin.mallardName1
import com.baishancloud.log.format.scala.Origin
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction
import org.apache.flink.util.Collector
import org.slf4j.{Logger, LoggerFactory}

import scala.collection.mutable

/**
 * 由于头条  域名数据从金华地区切换到石家庄地区 kafka 集群，因此抖音的请求需要两个地方进行写入，故在原来的基础上，在 tag 标签中增加 kafka_source 字段来群是消费的哪个地区的 kafka 集群数据。<br>
 *
 * @author ziqiang.wang 
 * @date 2022/3/1 16:01
 */
class LogParse1Sub(kafkaSource: String) extends BroadcastProcessFunction[Origin, mutable.Map[String, ServerCacheGroupEntity], Result1Sub] {

  private val logger: Logger = LoggerFactory.getLogger(classOf[LogParse1Sub])

  private var serverCacheGroupInfo: mutable.Map[String, ServerCacheGroupEntity] = mutable.Map[String, ServerCacheGroupEntity]()

  override def open(parameters: Configuration): Unit = {
    //获取外部资源
    ExternalDataFetchUtil.serverCacheGroupFetchFromFile(serverCacheGroupInfo)
  }

  override def processElement(value: Origin, ctx: BroadcastProcessFunction[Origin, mutable.Map[String, ServerCacheGroupEntity], Result1Sub]#ReadOnlyContext, out: Collector[Result1Sub]): Unit = {
    out.collect(
      Result1Sub(
        TimeUtil.timeAtLong(value.datetimeOfUTC().toEpochSecond, 60),
        60,
        mallardName1,
        "",
        1L,
        Tags1Sub(
          value.domain(),
          IpInfoUtil.core(value.ip()).isp,
          serverCacheGroupInfo.getOrElse(value.host(), ServerCacheGroupEntity.DEFAULT).cacheGroupCode,
          kafkaSource
        ),
        Fields1Sub(
          if (!StringUtil.split(value.complexField0(), '@').get(3).contains("dx-lt")) 1L else 0L
        )
      )
    )
  }

  override def processBroadcastElement(value: mutable.Map[String, ServerCacheGroupEntity], ctx: BroadcastProcessFunction[Origin, mutable.Map[String, ServerCacheGroupEntity], Result1Sub]#Context, out: Collector[Result1Sub]): Unit = {
    if (value.nonEmpty) {
      serverCacheGroupInfo = value
      logger.info(s"接收到广播数据，ipCacheGroupInfo表size为：${serverCacheGroupInfo.size}")
    }
  }
}
